package com.jthinking.logship.processor;


import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.RandomAccessFile;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;


public class LogsProcessor {

    private static final Logger LOGGER = LoggerFactory.getLogger(LogsProcessor.class);

    public static void process(File file, int singleFileThreads, int readBufferSize, Consumer<String> consumer) {

        long start = System.nanoTime();

        CountDownLatch latch = new CountDownLatch(singleFileThreads);

        long length = file.length();
        long blockSize = length / singleFileThreads;

        Map<Integer, Block> threadMap = new HashMap<>();

        for (int i = 0; i < singleFileThreads; i++) {
            long skip = blockSize * i;
            Block block = new Block();
            block.setStart(skip);
            block.setEnd(skip + blockSize + 1);
            threadMap.put(i, block);
        }

        AtomicLong atomicLong = new AtomicLong(0);
        for (Map.Entry<Integer, Block> entry : threadMap.entrySet()) {
            int i = entry.getKey();
            Block block = entry.getValue();

            Thread thread = new Thread(() -> {
                try (RandomAccessFile inputStream = new RandomAccessFile(file, "r");
                     ByteArrayOutputStream builder = new ByteArrayOutputStream()) {
                    long position = blockSize * i;
                    inputStream.seek(position);
                    if (block.getStart() != 0) {
                        int chr;
                        do {
                            position++;
                        } while ((chr = inputStream.read()) != '\n' && chr != -1);
                    }
                    byte[] temp = new byte[readBufferSize];
                    int len;
                    LOGGER.info("start" + i);
                    outer: while ((len = inputStream.read(temp, 0, readBufferSize)) != -1) {
                        for (int p = 0; p < len; p++) {
                            position++;
                            byte b = temp[p];
                            builder.write(b);
                            if (b == '\n') {
                                try {
                                    String line = builder.toString();
                                    consumer.accept(line);
                                    builder.reset();
                                    atomicLong.addAndGet(1);
                                    if (position >= block.getEnd()) {
                                        break outer;
                                    }
                                } catch (Throwable throwable) {
                                    LOGGER.error("", throwable);
                                }
                            }
                        }
                    }
                    if (len == -1) {
                        try {
                            String line = builder.toString();
                            consumer.accept(line);
                            builder.reset();
                            atomicLong.addAndGet(1);
                        } catch (Throwable throwable) {
                            LOGGER.error("", throwable);
                        }
                    }
                } catch (Exception e) {
                    LOGGER.error("", e);
                } finally {
                    latch.countDown();
                }
            });
            thread.setDaemon(true);
            thread.setName("logs-processor-thread-" + i);
            thread.start();
        }

        try {
            latch.await();
        } catch (InterruptedException e) {
            LOGGER.error("", e);
        }

        LOGGER.info("process size: {}", atomicLong.get());
        LOGGER.info("time cost: {}", System.nanoTime() - start);
    }

    private static class Block {
        private long start;
        private long end;

        public long getStart() {
            return start;
        }

        public void setStart(long start) {
            this.start = start;
        }

        public long getEnd() {
            return end;
        }

        public void setEnd(long end) {
            this.end = end;
        }
    }


}
